package com;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.mapred.*;

import java.util.List;
import java.util.Properties;

public class ORCReader {
    /**
     * https://blog.csdn.net/nysyxxg/article/details/52230399
     */
    public static void main(String[] args) throws Exception {
        JobConf conf = new JobConf();
        Path testFilePath = new Path("/home/hdfs/data/fly-11-1.orc");
        Properties p = new Properties();
        OrcSerde serde = new OrcSerde();
        p.setProperty("columns", "channel,time,params");
        p.setProperty("columns.types", "int:bigint:array<float>");
        serde.initialize(conf, p);
        StructObjectInspector inspector =
                (StructObjectInspector) ObjectInspectorFactory
                        .getReflectionObjectInspector(FlyParam.class, ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
//        StructObjectInspector inspector = (StructObjectInspector) serde.getObjectInspector();
        InputFormat in = new OrcInputFormat();
        FileInputFormat.setInputPaths(conf, testFilePath.toString());
        InputSplit[] splits = in.getSplits(conf, 1);
        System.out.println("splits.length==" + splits.length);
        conf.set("hive.io.file.readcolumn.ids", "1");
        RecordReader reader = in.getRecordReader(splits[0], conf, Reporter.NULL);
        Object key = reader.createKey();
        Object value = reader.createValue();
        List<? extends StructField> fields = inspector.getAllStructFieldRefs();
        long offset = reader.getPos();
        while (reader.next(key, value)) {
            Object channel = inspector.getStructFieldData(value, fields.get(0));
            Object time = inspector.getStructFieldData(value, fields.get(1));
            Object params = inspector.getStructFieldData(value, fields.get(2));
            offset = reader.getPos();
            System.out.println(channel + "|" + time + "|" + params);
        }
        reader.close();
    }

}
